package com.atguigu.utils;

import com.atguigu.common.ODSStaticConstants;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 */
public class KafkaUtil {
    // 默认topic，不会使用
    private static final String DEFAULT_TOPIC = "DWD_DEFAULT_TOPIC";
    public static SinkFunction<String> getKafkaProducter(String topic) {
        return new FlinkKafkaProducer<String>(
                ODSStaticConstants.KAFKA_BROKER_LIST,
                topic
                , new SimpleStringSchema());
    }

    public static <T> FlinkKafkaProducer<T> sendToDwdMessage(KafkaSerializationSchema<T> keyedSchema){
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,ODSStaticConstants.KAFKA_BROKER_LIST);
        return new FlinkKafkaProducer<T>(DEFAULT_TOPIC,
                keyedSchema,
                properties,
                // 如果不开启checkpoint，就算开启精准一次性也无用
                FlinkKafkaProducer.Semantic.NONE);
    }

    public static FlinkKafkaConsumer<String> getKafkaConsumer(String cdcToOdsKafkaTopicLog, String odsDwdGroupId) {
        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,ODSStaticConstants.KAFKA_BROKER_LIST);
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, odsDwdGroupId);
        return new FlinkKafkaConsumer<String>(cdcToOdsKafkaTopicLog,new SimpleStringSchema(),prop);
    }
}
